package net.xuzhen97.distributed.transaction.order.service.executor;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import net.xuzhen97.distributed.transaction.order.common.util.JacksonUtil;
import net.xuzhen97.distributed.transaction.order.dao.Order;
import net.xuzhen97.distributed.transaction.order.dao.OrderRepository;
import net.xuzhen97.distributed.transaction.order.dao.OrderState;
import net.xuzhen97.distributed.transaction.order.manager.microservice.CourseClient;
import net.xuzhen97.distributed.transaction.order.manager.microservice.OpenCourseCmd;
import net.xuzhen97.distributed.transaction.order.service.dto.OrderPayCompleteCmd;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.Objects;

@Service
@Slf4j
public class OrderPayCompleteCmdExe {

    private final OrderRepository orderRepository;
    private final RocketMQTemplate rocketMQTemplate;

    @Autowired
    public OrderPayCompleteCmdExe(OrderRepository orderRepository
            , RocketMQTemplate rocketMQTemplate){
        this.orderRepository = orderRepository;
        this.rocketMQTemplate = rocketMQTemplate;
    }

    @Transactional(rollbackFor = Exception.class)
    public String execute(OrderPayCompleteCmd cmd){
        cmd.validate();

        Order order = orderRepository.getById(cmd.getId());
        if(order !=null && Objects.equals(OrderState.UNPAY.name(), order.getState())){

            // 修改订单状态未支付状态
            order.setState(OrderState.PAY.name());
            order.setLastModifiedBy(cmd.getOpUserId());
            order.setLastModifiedDate(LocalDateTime.now());

            orderRepository.updateById(order);

            log.info("准备发送事件！！！");

            MessageParams messageParams = new MessageParams();
            messageParams.setCourseId(order.getCourseId());
            messageParams.setUserId(order.getUserId());
            messageParams.setOpUserId(cmd.getOpUserId());

            rocketMQTemplate.send("order"
                    , MessageBuilder.withPayload(messageParams).build());

            return order.getId();
        }
        return null;
    }

    /**
     * rocketmq事件消息实体
     * @author xuzhen97
     */
    @Data
    public static class MessageParams implements Serializable {
        private String userId;
        private String courseId;
        private String opUserId;
    }

    /**
     * rocketmq消费开课消息
     *  利用rocketmq的机制来存储消息和重试消息
     * @author xuzhen97
     */
    @Component
    @RocketMQMessageListener(topic = "order", consumerGroup = "order")
    @Slf4j
    public static class OrderPayCompleteConsumer implements RocketMQListener<MessageParams> {

        private final CourseClient courseClient;

        @Autowired
        public OrderPayCompleteConsumer(CourseClient courseClient) {
            this.courseClient = courseClient;
        }

        @Override
        public void onMessage(MessageParams params) {
            log.info(JacksonUtil.toJson(params));
            // 自动开课代码
            OpenCourseCmd openCourseCmd = new OpenCourseCmd();
            openCourseCmd.setUserId(params.getUserId());
            openCourseCmd.setCourseId(params.getCourseId());

            // 如果非200 到 300状态码会抛出异常，异常会使得我们的代码回滚
            courseClient.openCourse(params.getOpUserId(), openCourseCmd);
        }
    }

}
